/*
 * [y] hybris Platform
 *
 * Copyright (c) 2000-2017 SAP SE
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of SAP
 * Hybris ("Confidential Information"). You shall not disclose such
 * Confidential Information and shall use it only in accordance with the
 * terms of the license agreement you entered into with SAP Hybris.
 */
package acc.asset.sync.job;

import de.hybris.platform.core.model.media.MediaFolderModel;
import de.hybris.platform.core.model.media.MediaModel;
import de.hybris.platform.cronjob.enums.CronJobResult;
import de.hybris.platform.cronjob.enums.CronJobStatus;
import de.hybris.platform.cronjob.model.CronJobModel;
import de.hybris.platform.jalo.media.MediaManager;
import de.hybris.platform.servicelayer.cronjob.AbstractJobPerformable;
import de.hybris.platform.servicelayer.cronjob.PerformResult;
import de.hybris.platform.servicelayer.impex.ImpExResource;
import de.hybris.platform.servicelayer.impex.ImportConfig;
import de.hybris.platform.servicelayer.impex.ImportResult;
import de.hybris.platform.servicelayer.impex.ImportService;
import de.hybris.platform.servicelayer.impex.impl.StreamBasedImpExResource;
import de.hybris.platform.servicelayer.media.MediaService;
import de.hybris.platform.servicelayer.model.ModelService;
import de.hybris.platform.servicelayer.type.TypeService;
import de.hybris.platform.util.CSVConstants;
import de.hybris.platform.util.Config;
import de.hybris.y2ysync.model.media.SyncImpExMediaModel;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.UUID;

import org.apache.commons.lang3.SerializationUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.log4j.Logger;

import acc.asset.sync.bean.DemoSyncData;
import acc.asset.sync.bean.DemoSyncRequest;


/**
 *
 */
public class ConsumeJobPerformable extends AbstractJobPerformable<CronJobModel>
{
	private final static Logger LOG = Logger.getLogger(ConsumeJobPerformable.class);

	private ImportService importService;
	private ModelService modelService;
	private MediaService mediaService;
	private TypeService typeService;

	@Override
	public boolean isAbortable()
	{
		return true;
	}

	@Override
	public boolean isPerformable()
	{
		return true;
	}

	@Override
	public PerformResult perform(final CronJobModel var1)
	{
		@SuppressWarnings("resource")
		final KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(getProperties());
		consumer.subscribe(Arrays.asList(getTopic()));
		final String homeURL = getHomeUrl();

		try
		{
			while (true)
			{
				final ConsumerRecords<String, byte[]> records = consumer.poll(100);


				final StringBuilder builder = new StringBuilder();
				for (final ConsumerRecord<String, byte[]> record : records)
				{
					final DemoSyncRequest requestData = (DemoSyncRequest) SerializationUtils.deserialize(record.value());
					if (!homeURL.equals(requestData.getSourcePlatformUrl()))
					{
						final List<DemoSyncData> list = requestData.getDemoSyncDataList();
						list.forEach(demoSyncData -> {
							builder.append(demoSyncData.getImpexHeader()).append("\n");
							builder.append(demoSyncData.getContent());
						});
					} //end if
				} //end for

				importData(builder.toString());

			} //end while
		}
		catch (final Exception e)
		{
			LOG.info(e.getMessage(), e);
			return new PerformResult(CronJobResult.FAILURE, CronJobStatus.FINISHED);
		}
	}

	private Properties getProperties()
	{
		final Properties props = new Properties();
		props.put("bootstrap.servers", getKafkaBootstrapServers());

		final String groupID = Config.getString("kafka.consumer.group.id", "");
		final String keyDeserializer = Config.getString("kafka.consumer.key.deserializer", "");
		final String valueDeserialize = Config.getString("kafka.consumer.value.deserializer", "");

		props.put("group.id", groupID);
		props.put("enable.auto.commit", Boolean.valueOf(Config.getBoolean("kafka.consumer.enable.auto.commit", true)));
		props.put("auto.commit.interval.ms", Integer.valueOf(Config.getInt("kafka.consumer.auto.commit.interval.ms", 1000)));
		props.put("key.deserializer", keyDeserializer);
		props.put("value.deserializer", valueDeserialize);

		return props;
	}

	private String getTopic()
	{
		final String topic = Config.getString("kafka.consumer.topic", "");
		if (StringUtils.isBlank(topic))
		{
			throw new IllegalArgumentException("the value for kafka.consumer.topic can not be empty");
		}

		return topic;
	}


	private String getPathForMedia(final MediaModel media)
	{
		LOG.info(" start get file for media:" + media.getCode());
		final MediaFolderModel mediaFolder = folderWithFallbackToRoot(media.getFolder());
		final String location = media.getLocation();
		final File file = MediaManager.getInstance().getMediaAsFile(mediaFolder.getQualifier(), location);

		LOG.info("end get file for media:" + media.getCode());
		return file.getAbsolutePath();
	}

	public String getKafkaBootstrapServers()
	{
		final String servers = Config.getString("kafka.bootstrap.servers", "");
		if (StringUtils.isBlank(servers))
		{
			throw new IllegalArgumentException("the value for kafka.bootstrap.servers can not be empty");
		}

		return servers;
	}

	private String createMediaForScript(final DemoSyncData syncData)
	{
		LOG.info("create media start");
		final SyncImpExMediaModel media = (SyncImpExMediaModel) this.modelService.create(SyncImpExMediaModel.class);
		final String code = getTypeCode(syncData.getSoureSyncImpexMeidaCode()) + UUID.randomUUID().toString();
		media.setCode(code);
		media.setMime("text/plain");
		media.setRealFileName(code);
		media.setImpexHeader(syncData.getImpexHeader());
		media.setSyncType(this.typeService.getComposedTypeForCode(syncData.getSyncType()));
		media.setExportCronJob(null);
		media.setDataHubColumns("");
		media.setDataHubType("");
		this.modelService.save(media);
		this.mediaService.setDataForMedia(media, syncData.getContent().getBytes());

		LOG.info("create media end");
		return getPathForMedia(media);

	}

	private String getTypeCode(final String soureSyncImpexMeidaCode)
	{
		String typeCode = "";
		if (StringUtils.isNotBlank(soureSyncImpexMeidaCode))
		{
			typeCode = soureSyncImpexMeidaCode.split("-")[0];
		}
		else
		{
			//TODO something wei.f.zhang later
			typeCode = "sync";
		}

		return typeCode;

	}

	private void importData(final String script)
	{
		if (StringUtils.isBlank(script))
		{
			return;
		}

		try
		{
			LOG.info("scrript:" + script);
			final ImpExResource mediaRes = new StreamBasedImpExResource(new ByteArrayInputStream(script.getBytes()),
					CSVConstants.HYBRIS_ENCODING);

			final ImportConfig config = new ImportConfig();
			config.setScript(mediaRes);

			final ImportResult result = importService.importData(config);
			LOG.info("import result finished:" + result.isFinished() + " success:" + result.isSuccessful() + " running:"
					+ result.isRunning());

		}
		catch (final Exception e)
		{
			LOG.info(e.getMessage(), e);
		}
	}

	private MediaFolderModel folderWithFallbackToRoot(final MediaFolderModel folder)
	{
		return folder == null ? this.getMediaService().getRootFolder() : folder;
	}

	private String getHomeUrl()
	{
		return Config.getParameter("y2ysync.home.url");
	}


	/**
	 * @return the importService
	 */
	public ImportService getImportService()
	{
		return importService;
	}

	/**
	 * @param importService
	 *           the importService to set
	 */
	public void setImportService(final ImportService importService)
	{
		this.importService = importService;
	}

	/**
	 * @return the modelService
	 */
	public ModelService getModelService()
	{
		return modelService;
	}

	/**
	 * @param modelService
	 *           the modelService to set
	 */
	@Override
	public void setModelService(final ModelService modelService)
	{
		super.setModelService(modelService);
		this.modelService = modelService;
	}

	/**
	 * @return the mediaService
	 */
	public MediaService getMediaService()
	{
		return mediaService;
	}

	/**
	 * @param mediaService
	 *           the mediaService to set
	 */
	public void setMediaService(final MediaService mediaService)
	{
		this.mediaService = mediaService;
	}

	/**
	 * @return the typeService
	 */
	public TypeService getTypeService()
	{
		return typeService;
	}

	/**
	 * @param typeService
	 *           the typeService to set
	 */
	public void setTypeService(final TypeService typeService)
	{
		this.typeService = typeService;
	}
}
